package cloud.xlh.dynamic_env_refresh_server.service.impl;


import cloud.xlh.dynamic_env_refresh_server.NotifyCenter;
import cloud.xlh.dynamic_env_refresh_server.event.Event;
import cloud.xlh.dynamic_env_refresh_server.event.LocalDataChangeEvent;
import cloud.xlh.dynamic_env_refresh_server.event.Subscriber;
import cloud.xlh.dynamic_env_refresh_server.service.LongPollingService;
import cloud.xlh.dynamic_env_refresh_server.utils.MD5Util;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.servlet.AsyncContext;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.util.*;
import java.util.concurrent.*;

/**
 * Created with IntelliJ IDEA.
 *
 * @ project name : xlh_osms
 * @ Author: XuLeHuang
 * @ Date: 2023/3/10 11:47
 * @ Description:
 */
@Slf4j
@Service
public class LongPollingServiceImpl implements LongPollingService {
    public static final String LONG_POLLING_HEADER = "Long-Pulling-Timeout";
    public static final String LONG_POLLING_NO_HANG_UP_HEADER = "Long-Pulling-Timeout-No-Hangup";
    private static volatile Map<String, String> switches = new HashMap<String, String>();
    public static final String FIXED_DELAY_TIME = "fixedDelayTime";
    /**
     * ClientLongPolling subscibers.
     */
    public  Queue<ClientLongPolling> allSubs;

    private NotifyCenter notifyCenter = new NotifyCenter();
    public void addLongPollingClient(HttpServletRequest req, HttpServletResponse rsp, Map<String, String> clientMd5Map,
                                     int probeRequestSize) {
        String str = req.getHeader(LongPollingServiceImpl.LONG_POLLING_HEADER);
        String noHangUpFlag = req.getHeader(LongPollingServiceImpl.LONG_POLLING_NO_HANG_UP_HEADER);
        String appName = req.getHeader("Client-AppName");
        String tag = req.getHeader("Vipserver-Tag");
        int delayTime = getSwitchInteger(LongPollingServiceImpl.FIXED_DELAY_TIME, 500);
        // Add delay time for LoadBalance, and one response is returned 500 ms in advance to avoid client timeout.
        long timeout = Math.max(10000, Long.parseLong(str) - delayTime);
        /*if (isFixedPolling()) {
            timeout = Math.max(10000, getFixedPollingInterval());
            // Do nothing but set fix polling timeout.
        } else {
            long start = System.currentTimeMillis();
            List<String> changedGroups = MD5Util.compareMd5(req, rsp, clientMd5Map);
            if (changedGroups.size() > 0) {
                generateResponse(req, rsp, changedGroups);
                LogUtil.CLIENT_LOG.info("{}|{}|{}|{}|{}|{}|{}", System.currentTimeMillis() - start, "instant",
                        RequestUtil.getRemoteIp(req), "polling", clientMd5Map.size(), probeRequestSize,
                        changedGroups.size());
                return;
            } else if (noHangUpFlag != null && noHangUpFlag.equalsIgnoreCase(TRUE_STR)) {
                LogUtil.CLIENT_LOG.info("{}|{}|{}|{}|{}|{}|{}", System.currentTimeMillis() - start, "nohangup",
                        RequestUtil.getRemoteIp(req), "polling", clientMd5Map.size(), probeRequestSize,
                        changedGroups.size());
                return;
            }
        }*/
        //String ip = RequestUtil.getRemoteIp(req);

        //List<String> changedGroups = MD5Util.compareMd5(req, rsp, clientMd5Map);
        // Must be called by http thread, or send response.
        final AsyncContext asyncContext = req.startAsync();

        // AsyncContext.setTimeout() is incorrect, Control by oneself
        asyncContext.setTimeout(0L);

        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);

        executorService.execute(
                new ClientLongPolling(asyncContext, clientMd5Map, "127.0.0.1", probeRequestSize, timeout, appName, tag));
    }

    public static int getSwitchInteger(String key, int defaultValue) {
        int rtn;
        try {
            String status = switches.get(key);
            rtn = status != null ? Integer.parseInt(status) : defaultValue;
        } catch (Exception e) {
            rtn = defaultValue;
            log.error("corrupt switch value {}={}", key, switches.get(key));
        }
        return rtn;
    }

    public LongPollingServiceImpl() {
        allSubs = new ConcurrentLinkedQueue<ClientLongPolling>();
        // Register A Subscriber to subscribe LocalDataChangeEvent.
        notifyCenter.registerSubscriber(new Subscriber() {
            @Override
            public void onEvent(Event event) {
                    if (event instanceof LocalDataChangeEvent) {
                        LocalDataChangeEvent evt = (LocalDataChangeEvent) event;
                        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
                        executorService.execute(new DataChangeTask(evt.groupKey, evt.isBeta, evt.betaIps));
                    }
            }
            @Override
            public Class<? extends Event> subscribeType() {
                return LocalDataChangeEvent.class;
            }
        });
    }

    /**
     * DataChangeTask
     **/
    class DataChangeTask implements Runnable {

        @Override
        public void run() {
            try {
                //ConfigCacheService.getContentBetaMd5(groupKey);
                for (Iterator<ClientLongPolling> iter = allSubs.iterator(); iter.hasNext(); ) {
                    ClientLongPolling clientSub = iter.next();
                    if (clientSub.clientMd5Map.containsKey(groupKey)) {
                        // If published tag is not in the beta list, then it skipped.
                        /*if (isBeta && !CollectionUtils.contains(betaIps, clientSub.ip)) {
                            continue;
                        }*/

                        // If published tag is not in the tag list, then it skipped.
                       /* if (StringUtils.isNotBlank(tag) && !tag.equals(clientSub.tag)) {
                            continue;
                        }*/

                        //getRetainIps().put(clientSub.ip, System.currentTimeMillis());
                        iter.remove(); // Delete subscribers' relationships.
                        /*LogUtil.CLIENT_LOG
                                .info("{}|{}|{}|{}|{}|{}|{}", (System.currentTimeMillis() - changeTime), "in-advance",
                                        RequestUtil
                                                .getRemoteIp((HttpServletRequest) clientSub.asyncContext.getRequest()),
                                        "polling", clientSub.clientMd5Map.size(), clientSub.probeRequestSize, groupKey);*/
                        clientSub.sendResponse(Arrays.asList(groupKey));
                    }
                }

            } catch (Throwable t) {
                //LogUtil.DEFAULT_LOG.error("data change error: {}", ExceptionUtil.getStackTrace(t));
            }
        }

        DataChangeTask(String groupKey, boolean isBeta, List<String> betaIps) {
            this(groupKey, isBeta, betaIps, null);
        }

        DataChangeTask(String groupKey, boolean isBeta, List<String> betaIps, String tag) {
            this.groupKey = groupKey;
            this.isBeta = isBeta;
            this.betaIps = betaIps;
            this.tag = tag;
        }

        final String groupKey;

        final long changeTime = System.currentTimeMillis();

        final boolean isBeta;

        final List<String> betaIps;

        final String tag;

    }

    /**
     * ClientLongPolling
     **/
    public class ClientLongPolling implements Runnable {
        final long timeoutTime;
        final AsyncContext asyncContext;
        final Map<String, String> clientMd5Map;
        final long createTime;
        final String ip;
        final String appName;
        final String tag;
        final int probeRequestSize;
        Future<?> asyncTimeoutFuture;

        public ClientLongPolling(AsyncContext ac, Map<String, String> clientMd5Map, String ip, int probeRequestSize,
                                 long timeoutTime, String appName, String tag) {
            this.asyncContext = ac;
            this.clientMd5Map = clientMd5Map;
            this.probeRequestSize = probeRequestSize;
            this.createTime = System.currentTimeMillis();
            this.ip = ip;
            this.timeoutTime = timeoutTime;
            this.appName = appName;
            this.tag = tag;
        }

        @Override
        public void run() {
            ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);

            //29.5s就返回，无论在29.5s那一刻是不是检查出有变化
            executorService.schedule(new Runnable() {
                @Override
                public void run() {
                    try {
                        //getRetainIps().put(ClientLongPolling.this.ip, System.currentTimeMillis());
                        // Delete subscriber's relations.
                        boolean removeFlag = allSubs.remove(ClientLongPolling.this);
                        System.out.println(removeFlag);
                        if (removeFlag) {
                            List<String> changedGroups = MD5Util.compareMd5((HttpServletRequest) asyncContext.getRequest(),
                                    (HttpServletResponse) asyncContext.getResponse(), clientMd5Map);

                            if (changedGroups!=null && changedGroups.size() > 0) {
                                sendResponse(changedGroups);
                            } else {
                                sendResponse(null);
                            }
                        }
                    } catch (Throwable t) {
                        t.printStackTrace();
                    }
                }

            }, timeoutTime, TimeUnit.MILLISECONDS);
            allSubs.add(this);
        }

        void sendResponse(List<String> changedGroups) {
            // Cancel time out task.
            if (null != asyncTimeoutFuture) {
                asyncTimeoutFuture.cancel(false);
            }
            generateResponse(changedGroups);
        }

        void generateResponse(List<String> changedGroups) {
            if (null == changedGroups) {
                // Tell web container to send http response.
                asyncContext.complete();
                return;
            }

            HttpServletResponse response = (HttpServletResponse) asyncContext.getResponse();

            try {
                //final String respString = MD5Util.compareMd5ResultString(changedGroups);

                // Disable cache.
                response.setHeader("Pragma", "no-cache");
                response.setDateHeader("Expires", 0);
                response.setHeader("Cache-Control", "no-cache,no-store");
                response.setStatus(HttpServletResponse.SC_OK);
                System.out.println(changedGroups);
                response.getWriter().println(changedGroups);
                asyncContext.complete();
            } catch (Exception ex) {
                log.error(ex.toString(), ex);
                asyncContext.complete();
            }
        }

        public List<String> compareMd5(HttpServletRequest request, HttpServletResponse response,
                                              Map<String, String> clientMd5Map) {
            List<String> changedGroupKeys = new ArrayList<String>();
            String tag = request.getHeader("Vipserver-Tag");
            return changedGroupKeys;
        }
    }

    public NotifyCenter getNotifyCenter() {
        return this.notifyCenter;
    }
}
